import { useEffect, useState, useRef, useImperativeHandle, forwardRef } from 'react'
import { useNavigate } from 'react-router-dom'
import { Button, message, Modal, Flex, Tag, Input, Row, Col, Space, Table, Divider } from 'antd'
import { ModalForm, ProFormDigit, ProFormItem, ProFormSelect, ProFormText, ProFormTextArea, ProFormUploadButton } from '@ant-design/pro-components'
import { keepPreviousData, useQuery } from '@tanstack/react-query'
import SearchForm from "@/components/searchForm";
import { kafkaConnectDetail, kafkaTopicAdd, kafkaTopicBatchDelete, kafkaTopicPage, kafkaTopicPartitions, kafkaTopicPost } from '@/api/sharing/kafka'
import { parseTime } from '@/utils'
import StringEditor from './value/StringEditor'



//i18n
const KafkaRightPanel = ({ currentSource, onTransferLeftPanel }) => {
  const [fullData, setFullData] = useState({})
  const [queryParam, setQueryParam] = useState({ partition: 0 });
  const [selectedRowKeys, setSelectedRowKeys] = useState([])
  const [selectedRow, setSelectedRow] = useState(null)
  const topicInfoModalRef = useRef(null)
  const confirm = Modal.confirm

  const queryItems = [
    {
      name: 'partition',
      label: '分区',
      type: 'SELECT_DEBOUNCE',
      fieldProps: {
        disabled: !currentSource,
        showSearch: false,
        debounceTimeout: 0,
        fetchOptions: () => {
          return kafkaTopicPartitions(currentSource.key, { id: currentSource?.connectInfo?.id }).then(res => {
            return res.result?.map(d => ({ label: d.partition, value: d.partition }))
          })
        }
      }
    },
    {
      name: 'offset',
      label: '偏移',
      type: 'INPUT_NUMBER',
      fieldProps: {
        min: 0,
        precision: 0,
        style: {
          width: 100
        }
      }
    },
    {
      name: 'valueFilter',
      label: 'value',
      type: 'INPUT',
    }
  ];

  const { data: tableData, isFetching, refetch } = useQuery({
    queryKey: ['page-KafkaRightPanel-kafkaTopicPage', currentSource?.key, queryParam],
    queryFn: () => {
      return kafkaTopicPage(currentSource?.key, {
        id: currentSource?.connectInfo?.id,
        ...queryParam,
      }).then(({ result = [] }) => {
        result.forEach((item, index) => {
          item.id = (index + 1).toString()
        })
        return result
      })
    },
    placeholderData: keepPreviousData,
    enabled: !!currentSource,
  })

  const columns = [
    {
      title: '序号',
      dataIndex: 'index',
      width: 80,
      render: (text, record, index) => index + 1
    },
    {
      title: '分区',
      dataIndex: 'partition',
      width: 120
    },
    {
      title: '偏移',
      dataIndex: 'offset',
      width: 120
    },
    {
      title: 'key',
      dataIndex: 'key',
      width: 120
    },
    {
      title: '创建时间',
      dataIndex: 'timestamp',
      width: 120,
      render: (text) => {
        return parseTime(text)
      }
    },
    {
      title: 'value',
      dataIndex: 'value',
      width: 160,
      ellipsis: true
    },
  ];

  const onSelectChange = (selectedRowKeys, selectedRows) => {
    if (selectedRowKeys.length) {
      setSelectedRowKeys([selectedRowKeys[selectedRowKeys.length - 1]]);
      setSelectedRow({ ...selectedRows[selectedRowKeys.length - 1] })
    } else {
      setSelectedRow(null)
      setSelectedRowKeys([])
    }
  }

  const handleSetKeyTTL = (values) => {
    return setKeyTTL(values).then((res) => {
      if (res.code == 200) {
        message.success('设置TTL成功')
        setFullData({ ...fullData, ttl: values.ttl })
        return true
      }
    })
  }

  const handleSendTopic = (values) => {
    return kafkaTopicPost(values).then(res => {
      if (res.code === 200) {
        message.success('发送成功')
        refetch()
        return true
      }
    })
  }

  const handleAddTopic = (values) => {
    return kafkaTopicAdd(values).then(res => {
      if (res.code === 200) {
        message.success('新增成功')
        onTransferLeftPanel('toConnect')
        return true
      }
    })
  }

  const handleDeleteTopic = (topic) => {
    Modal.confirm({
      title: '确定删除Topic吗？',
      onOk: () => {
        return kafkaTopicBatchDelete(currentSource?.connectInfo?.id, [topic]).then(res => {
          if (res.code === 200) {
            message.success('删除成功')
            onTransferLeftPanel('toDelete')
            return true
          }
        })
      }
    })
  }

  return (
    <>
      <Row style={{ flex: 0, marginBottom: 20 }} gutter={[10, 10]}>
        <Col span={14}>
          <div><Input value={currentSource?.key} readOnly={true} /></div>
        </Col>
        <Col span={10}>
          <Space>
            <Button type="primary" disabled={!currentSource} onClick={() => topicInfoModalRef.current.open(currentSource)}>Topic信息</Button>
            <SendTopicBtn initialValues={{ id: currentSource?.connectInfo?.id, topic: currentSource?.key }} onFinish={handleSendTopic} />
            <AddTopicBtn initialValues={{ id: currentSource?.connectInfo?.id, brokerNum: currentSource?.brokerNum }} onFinish={handleAddTopic} />
            <Button disabled={!currentSource} danger onClick={() => handleDeleteTopic(currentSource?.key)}>删除Topic</Button>
          </Space>
        </Col>
      </Row>
      <SearchForm
        initialValues={queryParam}
        onSearch={(values) => {
          if (JSON.stringify(values) !== JSON.stringify(queryParam)) {
            setSelectedRowKeys([])
            setQueryParam({ ...values })
          } else {
            refetch()
          }
        }}
        items={queryItems}
      />
      <Table
        virtual
        scroll={{ y: 300 }}
        columns={columns}
        rowKey={(record) => record.id}
        loading={isFetching}
        dataSource={tableData || []}
        pagination={false}
        rowSelection={{ type: 'checkbox', columnWidth: 40, hideSelectAll: true, selectedRowKeys: selectedRowKeys, onChange: onSelectChange }}
        onRow={(record) => ({
          onClick: () => {
            if (selectedRowKeys.includes(record.id)) {
              setSelectedRow(null)
              setSelectedRowKeys([])
            } else {
              setSelectedRow({ ...record })
              setSelectedRowKeys([record.id])
            }
          }
        })}
      />

      <StringEditor readOnly={true} value={selectedRow?.value} onChange={(value) => setSelectedRow({
        ...(selectedRow || {}),
        value
      })} />

      <TopicInfoModal ref={topicInfoModalRef} />
    </>
  )
}

const SendTopicBtn = (props) => {
  return (
    <ModalForm
      width={'400px'}
      trigger={<Button disabled={!props.initialValues?.id} type="primary">发送Topic信息</Button>}
      title="发送Topic消息"
      modalProps={{ destroyOnClose: true }}
      onFinish={props.onFinish}
      grid={true}
      {...props}
    >
      <ProFormItem name="id" hidden />
      <ProFormText name="topic" label="topic" disabled placeholder="请输入" rules={[{ required: true, message: '不能为空' }]} />
      <ProFormSelect name="partition" label="partitions" fieldProps={{ min: 0, precision: 0 }} rules={[{ required: true, message: '不能为空' }]}
        request={() => {
          return kafkaTopicPartitions(props.initialValues?.topic, {
            id: props.initialValues?.id,
          }).then(({ result = [] }) => {
            return result.map(d => ({ label: d.partition, value: d.partition }))
          })
        }}
      />
      <ProFormText name="key" label="key" placeholder="请输入" rules={[{ required: true, message: '不能为空' }]} />
      <ProFormTextArea name="value" label="value" placeholder="请输入" rules={[{ required: true, message: '不能为空' }]} />
    </ModalForm>
  )
}

const AddTopicBtn = (props) => {
  return (
    <ModalForm
      width={'400px'}
      trigger={<Button disabled={!props.initialValues?.id} type="primary">新增Topic</Button>}
      title="新增topic"
      modalProps={{ destroyOnClose: true }}
      onFinish={props.onFinish}
      grid={true}
      {...props}
    >
      <ProFormItem name="id" hidden />
      <ProFormText name="name" label="topic" placeholder="请输入topic" rules={[{ required: true, message: '不能为空' }]} />
      <ProFormDigit name="numPartitions" label="分区数量" fieldProps={{ min: 1, precision: 0 }} rules={[{ required: true, message: '不能为空' }]} />
      <ProFormDigit name="replicationFactor" label="副本数量" fieldProps={{ min: 1, max: props.initialValues?.brokerNum, precision: 0 }} rules={[{ required: true, message: '不能为空' }]} />
    </ModalForm>
  )
}

const TopicInfoModal = forwardRef((props, ref) => {
  const [open, setOpen] = useState(false)
  const [topicInfo, setTopicInfo] = useState({})
  const [loading, setLoading] = useState(false)
  const columns = [
    {
      title: 'Partition',
      dataIndex: 'partition',
      width: 120
    },
    {
      title: 'Leader',
      dataIndex: 'leader',
      width: 120,
      render: (text, record) => {
        return record.leader.id
      }
    },
    {
      title: 'Beginning Offset',
      dataIndex: 'beginningOffset',
      width: 120
    },
    {
      title: 'End Offset',
      dataIndex: 'endOffset',
      width: 120
    },
    {
      title: 'Log Size',
      dataIndex: 'logSize',
      width: 120,
      render: (text, record) => {
        return record.replicas[0]?.logSize
      }
    },
    {
      title: 'Replicas',
      dataIndex: 'replicas',
      render: (text, record) => {
        return record.replicas.length
      },
      width: 120
    },
    {
      title: 'ISR',
      dataIndex: 'isr',
      render: (text, record) => {
        return record.isr[0]?.id
      },
      width: 120
    }
  ]
  useImperativeHandle(ref, () => ({
    open: (data) => {
      setOpen(true)
      fetchTopicInfo(data)
    }
  }))

  const fetchTopicInfo = (data) => {
    setLoading(true)
    kafkaTopicPartitions(data.key, {
      id: data.connectInfo?.id,
    }).then(({ result = [] }) => {
      setTopicInfo(result)
      setLoading(false)
    })
  }
  return (
    <Modal
      title="Topic信息"
      width={'70%'}
      okButtonProps={{ style: { display: 'none' } }}
      cancelButtonProps={{ style: { display: 'none' } }}
      loading={loading}
      open={open}
      onCancel={() => setOpen(false)}>
      <Table style={{ marginTop: 20 }} dataSource={topicInfo} columns={columns} pagination={false} bordered />
    </Modal>
  )
})

export default KafkaRightPanel
